package com.qingcloud.df.executor.log.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Snowflake;
import cn.hutool.core.util.IdUtil;
import cn.hutool.db.Entity;
import cn.hutool.db.GlobalDbConfig;
import cn.hutool.db.ds.simple.SimpleDataSource;
import cn.hutool.db.handler.EntityListHandler;
import cn.hutool.db.sql.SqlExecutor;
import com.qingcloud.df.executor.log.IComponentLogger;
import com.qingcloud.df.executor.pojo.JobInstanceLog;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.sql.Connection;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;

/**
 * 组件Crate数据库日志实现
 *
 * @author wangjinnan
 */
public class ComponentCrateLogger implements IComponentLogger {

    private static final Logger LOGGER = LoggerFactory.getLogger(ComponentCrateLogger.class);

    /**
     * 单例对象
     */
    private static volatile ComponentCrateLogger instance = null;

    /**
     * 运行节点ID
     */
    public static volatile Integer nodeId;

    /**
     * crate数据库地址
     */
    public static volatile String crateUrl;

    /**
     * crate数据库用户
     */
    public static volatile String crateUser;

    /**
     * crate数据库密码
     */
    public static volatile String cratePassword;

    /**
     * crate数据源
     */
    private DataSource dataSource;

    /**
     * 唯一ID生成工具
     */
    private Snowflake snowflake;

    static {
        //cratedb不支持返回主键模式
        GlobalDbConfig.setReturnGeneratedKey(false);
    }

    private ComponentCrateLogger() {

    }

    /**
     * 获取单例对象
     */
    public static ComponentCrateLogger getInstance() {
        if (Objects.isNull(instance)) {
            synchronized (ComponentCrateLogger.class) {
                if (Objects.isNull(instance)) {
                    instance = new ComponentCrateLogger();
                    instance.dataSource = new SimpleDataSource(crateUrl, crateUser, cratePassword,
                            "io.crate.client.jdbc.CrateDriver");
                }
            }
        }
        return instance;
    }

    /**
     * 获取生成日志记录ID
     */
    private long nextLogId() {
        if (Objects.isNull(snowflake)) {
            snowflake = IdUtil.getSnowflake(nodeId + 1, 6);
        }
        return snowflake.nextId();
    }

    /**
     * 追加utf8日志内容
     *
     * @param instanceId 组件实例ID
     * @param content    日志内容，utf8编码
     * @return 追加是否成功
     */
    @Override
    public Boolean appendUtf8Log(Integer jobId, Integer instanceId, String softwareName, String content, String env) {
        try (Connection connection = dataSource.getConnection()) {
            //日志其他属性
            long logId = nextLogId();
            final LocalDateTime now = LocalDateTime.now();
            String month = now.format(DateTimeFormatter.ofPattern("yyyyMM"));
            String time = now.format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss"));
            int size = SqlExecutor.execute(connection, "insert into df_component_runlog (id, instance_id, content, ins_month, ins_time) VALUES(?, ?, ?, ?, ?)",
                    logId, instanceId, content, month, time);
            if (size == 1) {
                return true;
            }
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }
        return false;
    }

    @Override
    public JobInstanceLog preLog(Integer jobId, Integer instanceId, String softwareName, Long preStart, Long afterStart, Integer size, String day, String env) {
        //拼接查询sql
        StringBuilder builder = new StringBuilder();
        builder.append("select id, content from df_component_runlog where instance_id = ? and ins_month=? and ins_time like ? ");
        if (preStart > 0) {
            //从某个ID往前查
            builder.append("and id < ").append(preStart).append(" ");
        }
        builder.append("order by ins_time desc limit ?");
        //查询日志
        try (Connection connection = dataSource.getConnection()) {
            //查询最新的size行日志
            List<Entity> entityList = SqlExecutor.query(connection, builder.toString(), new EntityListHandler(), instanceId, day.substring(0, 6), day + "%", size);
            return buildLogInstance(entityList, 2, preStart, afterStart);
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }
        return buildLogInstance(Collections.emptyList(), 2, preStart, afterStart);
    }

    @Override
    public JobInstanceLog afterLog(Integer jobId, Integer instanceId, String softwareName, Long preStart, Long afterStart, Integer size, String day, String env) {
        //拼接查询sql
        StringBuilder builder = new StringBuilder();
        builder.append("select id, content from df_component_runlog where instance_id = ? and ins_month=? and ins_time like ? ");
        if (afterStart > 0) {
            //从某个ID往前查
            builder.append("and id > ").append(afterStart).append(" ");
        }
        builder.append("order by ins_time asc limit ?");
        //查询日志
        try (Connection connection = dataSource.getConnection()) {
            //查询最新的size行日志
            List<Entity> entityList = SqlExecutor.query(connection, builder.toString(), new EntityListHandler(), instanceId, day.substring(0, 6), day + "%", size);
            return buildLogInstance(entityList, 1, preStart, afterStart);
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }
        return buildLogInstance(Collections.emptyList(), 1, preStart, afterStart);
    }

    @Override
    public JobInstanceLog readNewestLog(Integer jobId, Integer instanceId, String softwareName, Integer size, String day, String env) {
        try (Connection connection = dataSource.getConnection()) {
            //查询最新的size行日志
            String sql = "select id, content from df_component_runlog where instance_id = ? and ins_month=? and ins_time like ? order by ins_time desc limit ?";
            List<Entity> entityList = SqlExecutor.query(connection, sql, new EntityListHandler(), instanceId, day.substring(0, 6), day + "%", size);
            return buildLogInstance(entityList, 2, 0, 0);
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }
        return buildLogInstance(Collections.emptyList(), 2, 0, 0);
    }

    /**
     * 创建返回日志对象
     *
     * @param entityList 数据列表
     * @param orderType  排序类型 1正序， 2倒序
     * @return
     */
    private JobInstanceLog buildLogInstance(List<Entity> entityList, Integer orderType, long minId, long maxId) {
        //最大最小日志ID
        List<String> logList = Collections.emptyList();
        if (CollUtil.isNotEmpty(entityList)) {
            logList = new ArrayList<>(entityList.size());
            if (orderType == 1) {
                for (Entity entity : entityList) {
                    logList.add(entity.getStr("content"));
                    long id = entity.getLong("id");
                    if (id < minId) {
                        minId = id;
                    }
                    if (id > maxId) {
                        maxId = id;
                    }
                }
            } else {
                for (int i = entityList.size() - 1; i >= 0; i--) {
                    final Entity entity = entityList.get(i);
                    logList.add(entity.getStr("content"));
                    long id = entity.getLong("id");
                    if (id < minId) {
                        minId = id;
                    }
                    if (id > maxId) {
                        maxId = id;
                    }
                }
            }
        }
        final JobInstanceLog fileLog = new JobInstanceLog();
        fileLog.setPreStart(String.valueOf(minId));
        fileLog.setAfterStart(String.valueOf(maxId));
        fileLog.setLogs(logList);
        return fileLog;
    }

}
